package com.stan.core.spark.ad.blackList;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import com.stan.core.spark.ad.blackList.functions.*;
import com.google.common.base.Optional;
import com.stan.core.mapper.AdStreamBlackListMapper;
import com.stan.core.mapper.factory.MapperFactory;

import com.stan.core.vo.AdStreamBlackList;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import scala.Tuple2;

/**
 * 广告点击流黑名单过滤函数
 */
public class BlackListFilterFunction
        implements Function<JavaPairRDD<String, String>, JavaPairRDD<String, String>>,Serializable {

    private JavaSparkContext javaSparkContext;
    public BlackListFilterFunction(JavaSparkContext javaSparkContext) {
        this.javaSparkContext = javaSparkContext;
    }

    @Override
    public JavaPairRDD<String, String> call(
            // 格式: (224,2019-04-19 15:01:45 江苏 淮安 20 24)
            JavaPairRDD<String, String> adStreamRDD) throws Exception {
        // 1.从数据库查询被黑名单禁用的user,并封装成<userid,boolean>格式的RDD
        AdStreamBlackListMapper adStreamBlackListMapper = MapperFactory
                .getMapperFactory().getMapper(AdStreamBlackListMapper.class);
        List<AdStreamBlackList> AdStreamBlackListS = adStreamBlackListMapper
                .findAllBlackList();

        List<Tuple2<String,Boolean>> blackList = new ArrayList<Tuple2<String,Boolean>>();
        for(com.stan.core.vo.AdStreamBlackList AdStreamBlackList:AdStreamBlackListS){
            Tuple2<String,Boolean> userId2Bool = new Tuple2<String,Boolean>(AdStreamBlackList.getUserId(),true);
            blackList.add(userId2Bool);
        }

        JavaPairRDD<String,Boolean> userId2BoolRDD =
                javaSparkContext.parallelizePairs(blackList);
        // 2.将广告点击流转成(<userid,data>)进行左外连接
        // 注意这边不能使用内连接，内连接会导致join不上的左右数据全部丢失
        // 这边需要保证广告点击流的数据中不在黑名单中的数据不丢失

        // 首先将数据过滤一下，防止接下来运算中的错误
        adStreamRDD.filter(new SplitArrayFilterFunction());


        JavaPairRDD<String,String> userId2DataRDD = adStreamRDD.mapToPair(
              new StreamToUserId2DataRDDFunction());
        //进行左内连接，也就是黑名单中没有join上的数据可以丢弃，但是数据流中的不能
        JavaPairRDD<String,Tuple2<String, Optional<Boolean>>> userId2dataAndBlack =
                userId2DataRDD.leftOuterJoin(userId2BoolRDD);


        // 3.对Optional<Boolean>中取true的数据进行删除
        userId2dataAndBlack = userId2dataAndBlack.filter(new UserId2DataAndBlackFunction());

        // 4.转回(userId,data)格式，并进行返回
        JavaPairRDD<String,String> filteredUserId2DataRDD = userId2dataAndBlack.mapToPair(
                new UserId2DataAndBlackToUserId2DataFunction());

        return filteredUserId2DataRDD;
    }
}
